package com.shujia.streaming

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet, Statement}
import scala.collection.mutable.ListBuffer

object Demo4Bukong {
  def main(args: Array[String]): Unit = {
    // 缉查布控
    val conf: SparkConf = new SparkConf()
    conf.setMaster("local[2]") // 接收数据需要占用一个线程
    conf.setAppName("Demo1WordCountStreaming")
    val sc: SparkContext = new SparkContext(conf)
    // 创建StreamingContext的环境，设置处理时间间隔为5s
    // 相当于5s会封装成一个batch
    val ssc: StreamingContext = new StreamingContext(sc, Durations.seconds(5))
    val pointDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)
    // 算子外部的代码 默认在Driver端执行 只会执行一次
    //    val blackList: List[String] = get_blacklist()

    pointDS.filter(point => {
      // 直接放在算子内部 来一条数据执行一次 来一条数据就创建了一个连接
      val blackList: List[String] = get_blacklist()
      val mdn: String = point.split(",")(0)
      val bool: Boolean = blackList.contains(mdn)
      bool
    }).print()

    val filterDS: DStream[String] = pointDS
      // 实际上transform内的代码每个batch会执行一次
      // 将DS转换成RDD去操作
      .transform(rdd => {
        // 每5s创建一次连接
        val blackList: List[String] = get_blacklist()
        val rsc: SparkContext = rdd.sparkContext
        val blackListBro: Broadcast[List[String]] = rsc.broadcast(blackList)

        val filterRDD: RDD[String] = rdd.filter(point => {
          // 直接放在算子内部 来一条数据执行一次 来一条数据就创建了一个连接
          // val blackList: List[String] = get_blacklist()
          val mdn: String = point.split(",")(0)
          val bool: Boolean = blackListBro.value.contains(mdn)
          bool
        })
        filterRDD
      })

    filterDS.foreachRDD(rdd => {
      rdd.foreachPartition(rd => {
        Class.forName("com.mysql.jdbc.Driver")

        val conn: Connection = DriverManager.getConnection(
          "jdbc:mysql://master:3306/student"
          , "root"
          , "123456"
        )
        val stat: PreparedStatement = conn.prepareStatement("insert into result values(?)")
        rd.foreach(line => {
          stat.setString(1, line)
          stat.addBatch() // 使用批处理插入数据
        })
        stat.executeBatch()
        conn.close()

      })
    })



    // SparkStreaming程序需要手动启动
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }

  def get_blacklist(): List[String] = {

    val lb: ListBuffer[String] = ListBuffer[String]()
    // 从MySQL读取mdn黑名单数据
    Class.forName("com.mysql.jdbc.Driver")

    val conn: Connection = DriverManager.getConnection(
      "jdbc:mysql://master:3306/student"
      , "root"
      , "123456"
    )

    val state: Statement = conn.createStatement()
    val rs: ResultSet = state.executeQuery("select * from mdn_blacklist")
    while (rs.next()) {
      val mdn: String = rs.getString("mdn")
      lb.append(mdn)
    }
    conn.close()
    lb.toList
  }

}
